package com.ww.disruptor.consumer;

import com.lmax.disruptor.BusySpinWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.ww.disruptor.event.Trade;
import com.ww.disruptor.handler.*;
import com.ww.disruptor.producer.TradePublisher;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * TODO
 *
 * @author wanggw
 * @date 2024-09-14 23:41:29
 */
public class DisruptorDemo04 {

    public static void main(String[] args) throws InterruptedException {
        // 两个线程池
        // 构建用于提交的线程池
        ExecutorService es1 = Executors.newFixedThreadPool(4);
        // 用来发送消息的线程池
        ExecutorService es2 = Executors.newFixedThreadPool(5);

        // 构建Disruptor
        Disruptor<Trade> disruptor = new Disruptor<>(Trade::new,
                1024 * 1024,
                es2,
                ProducerType.SINGLE,
                new BusySpinWaitStrategy());
        // 串行
//        disruptor.handleEventsWith(new Handler1())
//                .handleEventsWith(new Handler2())
//                .handleEventsWith(new Handler3());

        // 并行
//        disruptor.handleEventsWith(new Handler1());
//        disruptor.handleEventsWith(new Handler2());
//        disruptor.handleEventsWith(new Handler3());

        // 菱形
        // Handler1跟Handler2是可以并行操作的，
        // 但是Handler3应该等待前两个完成后再执行才能正确获取到结果
//        disruptor.handleEventsWith(new Handler1(), new Handler2()).handleEventsWith(new Handler3());
//        disruptor.handleEventsWith(new Handler1(), new Handler2()).then(new Handler3());

        // 六边形操作
        Handler1 handler1 = new Handler1();
        Handler2 handler2 = new Handler2();
        Handler3 handler3 = new Handler3();
        Handler4 handler4 = new Handler4();
        Handler5 handler5 = new Handler5();

        disruptor.handleEventsWith(handler1, handler4);
        disruptor.after(handler1).handleEventsWith(handler2);
        disruptor.after(handler4).handleEventsWith(handler5);
        disruptor.after(handler2, handler5).handleEventsWith(handler3);


        // 启动Disruptor
        final RingBuffer<Trade> ringBuffer = disruptor.start();
        CountDownLatch latch = new CountDownLatch(1);
        long start = System.currentTimeMillis();
        es1.submit(new TradePublisher(latch, disruptor));

        latch.await();

        disruptor.shutdown();
        es1.shutdown();
        es2.shutdown();
        long end = System.currentTimeMillis();

        System.out.println("总耗时：" + (end - start));
    }
}
